package com.gizwits.kafkaActor

import java.util.Properties

import akka.actor.Actor
import akka.event.Logging
import com.gizwits.util.CaseClasssDomain.User
import com.gizwits.util._
import kafka.producer._
class ProductWorkerActor(producer:Producer[String, User]) extends  Actor{
  val props = new Properties()
  val log = Logging.getLogger(context.system, this)



  override def preStart(): Unit = {
    super.preStart()
    //do pre something
//    props ++= Map(
//      "serializer.class" -> Constants.SERIALIZEENCODERRCLASS,
//      "metadata.broker.list" -> Constants.BROKERS,
//      "key.serializer.class" ->"kafka.serializer.StringEncoder",
//      "producer.type"-> "async",
//      "compression.codec" -> "snappy"
//    )

  }

  override def receive: Receive = {
    case User(name:String,age:Int) =>{
      val u = User(name,age)
      log.info(u.toString)

       producer.send(new KeyedMessage[String, User](Constants.KafkaTopic, u))
    }
    case _=>{
      log.info("no macth")
    }
  }

  override def postStop(): Unit = {
    super.postStop()
    //stop somthing
  }
}
